package wangwenjun.phase3.executor.completablefuture.demo01;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import util.ThreadUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author ChangLiang
 * @date 2020/6/23
 */
@Slf4j
public class AppV1Test {

    /**
     * 场景1：一个请求需要向数据库中查询一批数据，然后对数据进行处理，最后统一输出结果
     */
    String[] mobiles = {"13233333330", "13233333331", "13233333332", "13233333333",
            "13233333334", "13233333335", "13233333336", "13233333337", "13233333338", "13233333339"};

    @BeforeEach
    public void setUp() {

    }

    @Test
    @DisplayName("test future#get")
    public void test() throws InterruptedException {
//        ThreadUtil.sleepForSeconds(5);

        log.warn("start");

        List<String> result = new ArrayList<>();

        // 其使用的是defaultThreadFactory 其thread是daemon=false
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        List<Callable<UserObject>> tasks = IntStream.range(0, 10).boxed()
                .map(i -> (Callable<UserObject>) () -> DbService.queryUserWithMobile(mobiles[i]))
                .collect(Collectors.toList());
        try {
            // invokeAll是同步方法
            executorService.invokeAll(tasks).stream().map(future -> {
                try {
                    // 第一阶段任务 从db中拉取一批数据 组成一个Stream<Integer>
                    return future.get();
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
            })
                    // 第二阶段任务 并行执行Stream<Integer> 对query进行处理
                    .parallel().forEach(dbQuery -> result.add(DbQueryHandler.handleUserDbQuery(dbQuery)));
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        log.warn("result:{}", result);
        // 这里join了 主线程没有结束 这个fixedThreadPool是不会shutdown的 但是ForkJoinPool是会shutdown的
//        Thread.currentThread().join();

        // question: 能不能第一阶段执行完成立即执行第二阶段任务呢？
    }

    @Test
    @DisplayName("test CompletableFuture")
    public void test2() throws InterruptedException {
        // for preparing jvisualvm
//        ThreadUtil.sleepForSeconds(5);
        log.warn("start");

        List<String> result = new ArrayList<>();

        /**
         * 生产环境下 这里应该定义10个CompletableFuture 执行allOf方法 通过get来阻塞获取同步结果
         */
        IntStream.range(0, 10).boxed()
                .forEach(i -> CompletableFuture.supplyAsync(() -> DbService.queryUserWithMobile(mobiles[i]))
                        .thenApply(DbQueryHandler::handleUserDbQuery)
                        .whenComplete((v, t) -> {
                            if (t == null) {
                                result.add(v);
                            } else {
                                log.error(ThreadUtil.exceptionToString(t));
                            }
                        })
                );
        log.info("finish immediately");
        // 有两个结果并没有在20秒内返回 就放弃掉
        ThreadUtil.sleepForSeconds(20);
        log.warn("result:{}", result);
        Thread.currentThread().join();
        // 仅仅使用ForkJoinPool ForkJoinPool是会shutdown的
    }

    @Test
    @DisplayName("test handleAsyncUse")
    public void test3() throws InterruptedException {
        // for preparing jvisualvm
//        ThreadUtil.sleepForSeconds(5);
        log.warn("start");

        List<String> result = new ArrayList<>();

        IntStream.range(0, 10).boxed()
                .forEach(i -> CompletableFuture.supplyAsync(() -> DbService.queryUserWithMobile(mobiles[i]))
                        .handleAsync((s,t)->{
                            Optional.of(t).ifPresent(e->log.error(ThreadUtil.exceptionToString(e)));
                            if (s != null) {
                                return DbQueryHandler.handleUserDbQuery(s);
                            }else{
                                return "";
                            }
                        })
                        .whenComplete((v, t) -> {
                            if (t == null) {
                                result.add(v);
                            } else {
                                log.error(ThreadUtil.exceptionToString(t));
                            }
                        })
                );
        log.info("finish immediately");
        // 有两个结果并没有在20秒内返回 就放弃掉
        ThreadUtil.sleepForSeconds(20);
        log.warn("result:{}", result);
        Thread.currentThread().join();
        // 仅仅使用ForkJoinPool ForkJoinPool是会shutdown的
    }

    @Test
    @DisplayName("test thenCompose")
    public void test4() throws InterruptedException {

        /**
         * thenCompose 两阶段使用同一个线程进行执行
         */

        log.warn("start");

        List<String> result = new ArrayList<>();

        IntStream.range(0, 10).boxed()
                .forEach(i -> CompletableFuture.supplyAsync(() -> DbService.queryUserWithMobile(mobiles[i]))
                        .thenCompose(user->CompletableFuture.supplyAsync(()->DbQueryHandler.handleUserDbQuery(user)))
                        .whenComplete((v, t) -> {
                            if (t == null) {
                                result.add(v);
                            } else {
                                log.error(ThreadUtil.exceptionToString(t));
                            }
                        })
                );
        log.info("finish immediately");
        // 有两个结果并没有在20秒内返回 就放弃掉
        ThreadUtil.sleepForSeconds(20);
        log.warn("result:{}", result);
        Thread.currentThread().join();
        // 仅仅使用ForkJoinPool ForkJoinPool是会shutdown的
    }

    @Test
    @DisplayName("test thenComposeAsync")
    public void test5() throws InterruptedException {

        /**
         * thenCompose 两阶段不一定使用同一个线程进行执行
         */

        log.warn("start");

        List<String> result = new ArrayList<>();

        IntStream.range(0, 10).boxed()
                .forEach(i -> CompletableFuture.supplyAsync(() -> DbService.queryUserWithMobile(mobiles[i]))
                        .thenComposeAsync(user->CompletableFuture.supplyAsync(()->DbQueryHandler.handleUserDbQuery(user)))
                        .whenComplete((v, t) -> {
                            if (t == null) {
                                result.add(v);
                            } else {
                                log.error(ThreadUtil.exceptionToString(t));
                            }
                        })
                );
        log.info("finish immediately");
        // 有两个结果并没有在20秒内返回 就放弃掉
        ThreadUtil.sleepForSeconds(20);
        log.warn("result:{}", result);
        Thread.currentThread().join();
        // 仅仅使用ForkJoinPool ForkJoinPool是会shutdown的
    }
}
